package com.ferry.lock.engine.zookeeper.menagerie.locks;

import com.ferry.lock.engine.LockHolder;
import com.ferry.lock.engine.zookeeper.menagerie.ZKSessionManager;
import com.ferry.lock.engine.zookeeper.menagerie.ZKUtils;
import com.ferry.lock.engine.zookeeper.menagerie.ZkPrimitive;
import com.ferry.lock.lock.IReentrantLock;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;

/**
 * Created by daiyong
 */
public class ReentrantZKLock extends ZkPrimitive implements IReentrantLock {

	private static final String lockPrefix = "lock";

	protected static final char lockDelimiter = '-';

	private final ThreadLocal<LockHolder> locks = new ThreadLocal<LockHolder>();

	public ReentrantZKLock(String baseNode, ZKSessionManager zkSessionManager) {
		super(baseNode, zkSessionManager, ZooDefs.Ids.OPEN_ACL_UNSAFE);
	}

	public ReentrantZKLock(String baseNode, ZKSessionManager zkSessionManager, List<ACL> privileges) {
		super(baseNode, zkSessionManager, privileges);
	}

	@Override
	public void lock() {
		/**
		 * 如果是当前线程重入 则直接返回
		 */
		if (checkReentrancy()) {
			return;
		}

		//设置监听
		setConnectionListener();
		//获取zk实例
		ZooKeeper zk = zkSessionManager.getZooKeeper();

		String lockNode;

		try {
			lockNode = zk.create(getBaseLockPath(), emptyNode, privileges, CreateMode.EPHEMERAL_SEQUENTIAL);
			while (true) {
				if (broken) {
					throw new RuntimeException("zk session expired, release lock");
				}

				localLock.lock();
				try {
					boolean acquiredLock = tryAcquireDistributed(zk, lockNode, true);
					if (!acquiredLock) {
						//获取不到锁，线程在此等待，等待watcher 解除等待
						condition.awaitUninterruptibly();
					} else {
						locks.set(new LockHolder(lockNode));
					}
				} finally {
					localLock.unlock();
				}
			}
		} catch (KeeperException e) {
			throw new RuntimeException(e);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		} finally {
			removeConnectionListener();
		}
	}

	@Override
	public void lockInterruptibly() throws InterruptedException {
		tryLock(Long.MAX_VALUE, TimeUnit.DAYS);
	}

	@Override
	public boolean tryLock() {
		//重入
		if (checkReentrancy()) return true;

		ZooKeeper zk = zkSessionManager.getZooKeeper();

		String lockNode = null;
		try {
			lockNode = zk.create(getBaseLockPath(), emptyNode, privileges, CreateMode.EPHEMERAL_SEQUENTIAL);

			boolean lockAcquired = tryAcquireDistributed(zk, lockNode, false);

			if (!lockAcquired) {
				//无需等待，删除当前节点 直接返回
				zk.delete(lockNode,-1);
				return false;
			}
			//获取成功
			else {
				locks.set(new LockHolder(lockNode));
				return true;
			}

		} catch (KeeperException e) {
			throw new RuntimeException(e);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {

		if (Thread.interrupted()) {
			throw new InterruptedException();
		}

		if (checkReentrancy()) return true;

		ZooKeeper zk = zkSessionManager.getZooKeeper();

		setConnectionListener();
		String lockNode;

		try {
			lockNode = zk.create(getBaseLockPath(), emptyNode, privileges, CreateMode.EPHEMERAL_SEQUENTIAL);

			while (true) {
				if (Thread.interrupted()) {
					zk.delete(lockNode, 1);
					throw new InterruptedException();
				} else if (broken) {
					throw new InterruptedException("zk session 过期");
				}

				boolean acquiredLock = tryAcquireDistributed(zk, lockNode, true);
				if(!acquiredLock){
					boolean alerted = condition.await(time, unit);
					if(!alerted){
						zk.delete(lockNode,-1);
						return false;
					}
				} else {
					locks.set(new LockHolder(lockNode));
					return true;
				}
			}

		} catch (KeeperException e) {
			throw new RuntimeException(e);
		} finally {
			removeConnectionListener();
		}
	}

	@Override
	public void unLock() {
		LockHolder nodeToRemove = locks.get();
		if(nodeToRemove==null)
			throw new IllegalMonitorStateException("Attempting to unlock without first obtaining that lock on this thread");

		int numLocks = nodeToRemove.decLockNum();
		if(numLocks==0){
			locks.remove();
			try {
				ZKUtils.safeDelete(zkSessionManager.getZooKeeper(), nodeToRemove.getLockNode(), -1);
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			} catch (KeeperException e) {
				throw new RuntimeException(e);
			}
		}
	}

	@Override
	public String getName() {
		return baseNode;
	}

	public Condition newCondition() {
		return new ZKCondition(baseNode, zkSessionManager, privileges, this);
	}


	private boolean checkReentrancy() {
		LockHolder local = locks.get();
		if(local != null){
			local.incLockNum();
			return true;
		}
		return false;
	}

	protected String getBaseLockPath(){
		return baseNode + "/" + getLockPrefix() + lockDelimiter;
	}

	protected String getLockPrefix() {
		return lockPrefix;
	}

	/**
	 * 获取分布式锁  排序后 依次向前找节点 如果不存在则获取锁 否则获取不到
	 * @param zk
	 * @param lockNode
	 * @param watch
	 * @return
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	protected boolean tryAcquireDistributed(ZooKeeper zk, String lockNode, boolean watch) throws KeeperException, InterruptedException {
		List<String> locks = ZKUtils.filterByPrefix(zk.getChildren(baseNode, false), getLockPrefix());
		ZKUtils.sortBySequence(locks, lockDelimiter);

		String myNodeName = lockNode.substring(lockNode.lastIndexOf('/') + 1);
		int myPos = locks.indexOf(myNodeName);

		int nextNodePos = myPos-1;
		while(nextNodePos>=0){
			Stat stat;
			if(watch)
				stat = zk.exists(baseNode + "/" + locks.get(nextNodePos), signalWatcher);
			else
				stat = zk.exists(baseNode + "/" + locks.get(nextNodePos), false);
			if (stat != null){
				//there is a node which already has the lock, so we need to wait for notification that that
				//node is gone
				return false;
			} else {
				nextNodePos--;
			}
		}

		return true;
	}

	public final boolean hasLock(){
		return locks.get() != null;
	}

	protected final String getLockName(){
		LockHolder lockHolder = locks.get();
		if(lockHolder == null) return null;
		return lockHolder.getLockNode();
	}
}
